package de.mmis.core.tuplespace;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.TimeoutException;

import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.LeafNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.event.Observable;
import de.mmis.core.base.event.Observer;
import de.mmis.core.tuplespace.Notification.Command;

/**
 * queue implementation using tuplespace to store elements
 * 
 * @author Christian Koch
 * 
 */
public class TuplespaceQueue extends TuplespaceContainer
		implements
			Queue<Tree>,
			Observable<TuplespaceEvent> {
	private static final String QUEUE_ELEM_NAME_SUFFIX = "-queue-elem";
	private static final String QUEUE_NAME_SUFFIX = "-queue";
	protected static final String QUEUE_NAME_PREFIX = ":";

	protected final Tuplespace space;
	protected final String name;
	protected final Tree id;
	protected final LeafNode queueName;
	protected final LeafNode queueElemName;
	protected final InnerNode queueQueryTuple;
	protected final InnerNode queueElemQueryTuple;
	private long lastReadElem;

	private final ArrayList<Observer<TuplespaceEvent>> observers;

	/**
	 * creates new tuplespace queue
	 * 
	 * @param space
	 *            tuplespace client connected to used server
	 * @param name
	 *            queue name (combined name and id should be unique)
	 * @param id
	 *            object id queue belongs to (combined name and id should be
	 *            unique)
	 * @throws TuplespaceException
	 *             when notification of queue operation fails
	 */
	public TuplespaceQueue(Tuplespace space, String name, Tree id)
			throws TuplespaceException {
		this(space, name, id, DEFAULT_TIMEOUT, DEFAULT_LEASETIME);
	}

	/**
	 * creates new tuplespace queue, that destroy itself when not used for a
	 * specific time
	 * 
	 * @param space
	 *            tuplespace client connected to used server
	 * @param name
	 *            queue name (combined name and id should be unique)
	 * @param id
	 *            object id queue belongs to (combined name and id should be
	 *            unique)
	 * @param timeout
	 *            default time span to wait trying to take queue tuple
	 * @throws TuplespaceException
	 *             when notification of queue operation fails
	 */
	public TuplespaceQueue(Tuplespace space, String name, Tree id,
			long timeout) throws TuplespaceException {
		this(space, name, id, timeout, DEFAULT_LEASETIME);
	}

	/**
	 * creates new tuplespace queue, that destroy itself when not used for a
	 * specific time
	 * 
	 * @param space
	 *            tuplespace client connected to used server
	 * @param name
	 *            queue name (combined name and id should be unique)
	 * @param id
	 *            object id queue belongs to (combined name and id should be
	 *            unique)
	 * @param timeout
	 *            default time span to wait trying to take queue tuple
	 * @param leaseTime
	 *            time to live between two consecutive push or pop operations
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public TuplespaceQueue(Tuplespace space, String name, Tree id,
			long timeout, long leaseTime) throws TuplespaceException {
		super(timeout, leaseTime);
		this.space = space;

		this.name = name;
		this.id = id;
		this.queueName = new LeafNode(
				(QUEUE_NAME_PREFIX + name + QUEUE_NAME_SUFFIX));
		this.queueElemName = new LeafNode(
				(QUEUE_NAME_PREFIX + name + QUEUE_ELEM_NAME_SUFFIX));

		this.queueQueryTuple = new InnerNode(queueName, id, new LeafNode("?"),
				new LeafNode("?"));
		this.queueElemQueryTuple = new InnerNode(queueElemName, id, new LeafNode("?"),
				new LeafNode("?"));

		observers = new ArrayList<Observer<TuplespaceEvent>>();

		Notification notification = new Notification(Command.WRITE,
				queueQueryTuple);
		space.notify(notification);

		space.addObserver(new Observer<TuplespaceEvent>() {
			@Override
			public void notify(Observable<? extends TuplespaceEvent> sender,
					TuplespaceEvent event) {
				if (queueQueryTuple.equals(event.getTemplate()))
					fireEvent(event);
			}
		});
	}

	/**
	 * (re)initializes the queue tuple, deleting all elements (not needed if
	 * already exists, needs to be done before any queue operations can be
	 * executed)
	 * 
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public void initQueue() throws TuplespaceException {
		deinitQueue(0);

		InnerNode queueTuple = new InnerNode(queueName, id, new LeafNode("1"), new LeafNode(
				"0"));

		space.write(queueTuple, leaseTime);
	}

	/**
	 * checks if queue tuple is present on server, using default timeout
	 * 
	 * @return true, if queue tuple is present, else false
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public boolean isPresent() throws TuplespaceException {
		return isPresent(getTimeout());
	}
	/**
	 * checks if queue tuple is present on server
	 * 
	 * @param timeout
	 *            time span to wait for queue tuple to appear. If set to 0, a
	 *            single readIfExists check will be made.
	 * @return true, if queue tuple is present, else false
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public boolean isPresent(long timeout) throws TuplespaceException {
		if (timeout <= 0)
			return space.readIfExists(queueQueryTuple) != null;
		if (space.read(queueQueryTuple, timeout) == null)
			return false;
		return true;
	}

	/**
	 * deletes the queue tuple and all elements, using default timeout
	 * 
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public void deinitQueue() throws TuplespaceException {
		deinitQueue(getTimeout());
	}
	/**
	 * deletes the queue tuple and all elements
	 * 
	 * @param timeout
	 *            time span to wait trying to remove queue tuple. If set to 0, a
	 *            single takeIfExists will be made
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 */
	public void deinitQueue(long timeout) throws TuplespaceException {
		if (timeout <= 0)
			space.takeIfExists(queueQueryTuple);
		else
			space.take(queueQueryTuple, timeout);

		while (space.takeIfExists(queueElemQueryTuple) != null) {
			// take until empty
		}
	}

	/**
	 * adds new element to the end of the queue with default leaseTime and
	 * timeout
	 * 
	 * @param tuple
	 *            element to add
	 * @return element index (may be used by derived classes)
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue tuple
	 */
	public long push(Tree... tuple) throws TuplespaceException,
			TimeoutException {
		return push(getTimeout(), leaseTime, tuple);
	}
	/**
	 * adds new element to the end of the queue with default leaseTime
	 * 
	 * @param tuple
	 *            element to add
	 * @param timeout
	 *            time span to wait trying to take queue tuple
	 * @return element index (may be used by derived classes)
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue tuple
	 */
	public long push(long timeout, Tree... tuple)
			throws TuplespaceException, TimeoutException {
		return push(timeout, leaseTime, tuple);
	}
	/**
	 * adds new element to the end of the queue with specified leaseTime
	 * 
	 * @param tuple
	 *            element to add
	 * @param timeout
	 *            time span to wait trying to take queue tuple
	 * @param leaseTime
	 *            time to live for inserted tuple
	 * @return element index (may be used by derived classes)
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue tuple
	 */
	public long push(long timeout, long leaseTime, Tree... tuple)
			throws TuplespaceException, TimeoutException {
		InnerNode queue = (InnerNode) space.take(queueQueryTuple, timeout);
		if (queue == null)
			throw new TimeoutException("queue take timed out");

		long end = Long.parseLong(((LeafNode) queue.getSubTree(3))
				.getUTF8Data());

		for(Tree exp : tuple)
		{
			end++;
			InnerNode temp = new InnerNode(queueElemName, id, new LeafNode(Long.toString(end)), exp);
			space.write(temp, leaseTime);
		}

		((LeafNode) queue.getSubTree(3)).setUTF8Data(Long.toString(end));
		space.write(queue, this.leaseTime);

		return end;
	}

	/**
	 * removes first element in queue, using default timeout
	 * 
	 * @return the removed first element
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public Tree pop() throws TuplespaceException, TimeoutException {
		return pop(getTimeout());
	}
	/**
	 * removes first element in queue
	 * 
	 * @param timeout
	 *            time span to wait trying to take queue tuple
	 * @return the removed first element
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public Tree pop(long timeout) throws TuplespaceException,
			TimeoutException {
		InnerNode queue = (InnerNode) space.take(queueQueryTuple, timeout);
		if (queue == null)
			throw new TimeoutException("queue take timed out");

		long start = Long.parseLong(((LeafNode) (queue).getSubTree(2))
				.getUTF8Data());
		long end = Long.parseLong(((LeafNode) (queue).getSubTree(3))
				.getUTF8Data());

		InnerNode temp = new InnerNode(queueElemName, id, new LeafNode(Long
				.toString(start)), new LeafNode("?"));
		InnerNode elem = null;
		do {
			if (start > end)
				break;

			((LeafNode) temp.getSubTree(2)).setUTF8Data(Long.toString(start));
			elem = (InnerNode) space.takeIfExists(temp);

			start++;
		} while (elem == null);

		((LeafNode) queue.getSubTree(2)).setUTF8Data(Long.toString(start));
		space.write(queue, leaseTime);

		if (elem == null)
			return null;

		lastReadElem = start - 1;
		return elem.getSubTree(3);
	}

	/**
	 * returns (but not removes) first element in queue, using default timeout
	 * 
	 * @return first element
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public Tree top() throws TuplespaceException, TimeoutException {
		return top(getTimeout());
	}
	/**
	 * returns (but not removes) first element in queue
	 * 
	 * @param timeout
	 *            time span to wait trying to read queue tuple
	 * @return first element
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public Tree top(long timeout) throws TuplespaceException,
			TimeoutException {
		InnerNode queue = (InnerNode) space.read(queueQueryTuple, timeout);
		if (queue == null)
			throw new TimeoutException("queue read timed out");

		long start = Long.parseLong(((LeafNode) (queue).getSubTree(2))
				.getUTF8Data());
		long end = Long.parseLong(((LeafNode) (queue).getSubTree(3))
				.getUTF8Data());

		InnerNode temp = new InnerNode(queueElemName, id, new LeafNode(Long
				.toString(start)), new LeafNode("?"));
		InnerNode elem = null;
		do {
			if (start > end)
				return null;

			((LeafNode) temp.getSubTree(2)).setUTF8Data(Long.toString(start));
			elem = (InnerNode) space.readIfExists(temp);

			if (elem == null)
				start++;
		} while (elem == null);

		lastReadElem = start;
		return elem.getSubTree(3);
	}

	/**
	 * @return number of elements in queue
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public long getSize() throws TuplespaceException, TimeoutException {
		return getSize(getTimeout());
	}

	/**
	 * @param timeout
	 *            time span to wait trying to read queue tuple
	 * @return number of elements in queue
	 * @throws TuplespaceException
	 *             error while reading or writing to tuplespace server
	 * @throws TimeoutException
	 *             timed out while trying to take queue or element tuples
	 */
	public long getSize(long timeout) throws TuplespaceException,
			TimeoutException {
		InnerNode queue = (InnerNode) space.read(queueQueryTuple, timeout);
		if (queue == null)
			throw new TimeoutException("queue read timed out");

		long start = Long.parseLong(((LeafNode) (queue).getSubTree(2))
				.getUTF8Data());
		long end = Long.parseLong(((LeafNode) (queue).getSubTree(3))
				.getUTF8Data());

		return (start > end) ? 0 : end - start + 1;
	}

	@Override
	public boolean add(Tree arg0) {
		try {
			push(getTimeout(), new Tree[]{arg0});
		} catch (TuplespaceException e) {
			throw new IllegalStateException(e);
		} catch (TimeoutException e) {
			throw new IllegalStateException(e);
		}
		return true;
	}

	@Override
	public Tree element() {
		Tree ret = peek();
		if (ret == null)
			throw new NoSuchElementException();
		return ret;
	}

	@Override
	public boolean offer(Tree arg0) {
		return add(arg0);
	}

	@Override
	public Tree peek() {
		try {
			return top();
		} catch (TuplespaceException e) {
			throw new IllegalStateException(e);
		} catch (TimeoutException e) {
			throw new IllegalStateException(e);
		}
	}

	@Override
	public Tree poll() {
		try {
			return pop();
		} catch (TuplespaceException e) {
			throw new IllegalStateException(e);
		} catch (TimeoutException e) {
			throw new IllegalStateException(e);
		}
	}

	@Override
	public Tree remove() {
		Tree ret = poll();
		if (ret == null)
			throw new NoSuchElementException();
		return ret;
	}

	@Override
	public boolean addAll(Collection<? extends Tree> arg0) {
		if (arg0.isEmpty())
			return false;
		for (Tree exp : arg0)
			add(exp);
		return true;
	}

	@Override
	public void clear() {
		try {
			initQueue();
		} catch (TuplespaceException e) {
			throw new UnsupportedOperationException(e);
		}
	}

	@Override
	public boolean contains(Object arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public boolean containsAll(Collection<?> arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public boolean isEmpty() {
		try {
			return getSize() == 0;
		} catch (TuplespaceException e) {
			throw new IllegalStateException(e);
		} catch (TimeoutException e) {
			throw new IllegalStateException(e);
		}
	}

	@Override
	public Iterator<Tree> iterator() {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public boolean remove(Object arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public boolean removeAll(Collection<?> arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public boolean retainAll(Collection<?> arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public int size() {
		try {
			return (int) getSize();
		} catch (TuplespaceException e) {
			throw new IllegalStateException(e);
		} catch (TimeoutException e) {
			throw new IllegalStateException(e);
		}
	}

	@Override
	public Object[] toArray() {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	@Override
	public <T> T[] toArray(T[] arg0) {
		throw new UnsupportedOperationException(
				"not supported by TuplespaceQueue");
	}

	/**
	 * @return name of queue
	 */
	public String getName() {
		return name;
	}

	/**
	 * @return id of object belonging to this queue
	 */
	public Tree getId() {
		return id;
	}

	/**
	 * @return index of last element read or taken
	 */
	public long getLastReadElem() {
		return lastReadElem;
	}

	protected void fireEvent(TuplespaceEvent event) {
		synchronized (observers) {
			for (Observer<TuplespaceEvent> observer : observers) {
				observer.notify(this, event);
			}
		}
	}

	@Override
	public void addObserver(Observer<TuplespaceEvent> observer) {
		synchronized (observers) {
			observers.add(observer);
		}
	}

	@Override
	public void removeObserver(Observer<TuplespaceEvent> observer) {
		synchronized (observers) {
			observers.remove(observer);
		}
	}
}
